package link.jfire.jnet.client;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import link.jfire.jnet.common.InternalResultContainer;
import link.jfire.jnet.common.buffer.ByteBuf;
import link.jfire.jnet.common.buffer.DirectByteBuf;
import link.jfire.jnet.common.buffer.DirectByteBufPool;
import link.jfire.jnet.common.decodec.FrameDecodec;
import link.jfire.jnet.common.exception.SocketException;
import link.jfire.jnet.common.handler.DataHandler;

/**
 * 客户端工具类，注意，该客户端是非线程安全类，其方法不可多线程运行
 * 
 * @author linbin
 * 
 */
public abstract class AioClient
{
	protected long						readTimeout		= 3000;
	// 默认的超时等待时间是30分钟
	protected long						waitTimeout		= 1000 * 60 * 30;
	protected String					address;
	protected int						port;
	protected AsynchronousSocketChannel	socketChannel;
	protected AsynchronousChannelGroup	channelGroup;
	protected FrameDecodec				frameDecodec;
	protected DataHandler[]				readHandlers;
	protected DataHandler[]				writeHandlers;
	protected DirectByteBuf				ioBuf			= DirectByteBufPool.getInstance().get(100);
	protected ClientReadCompleter		readCompleter;
	protected volatile boolean			connectState	= UNCONNECTED;
	public static final boolean			UNCONNECTED		= false;
	public static final boolean			CONNECTED		= true;
	protected Future<Void>				NORESULT		= new Future<Void>() {
		                                                    
		                                                    @Override
		                                                    public boolean cancel(boolean mayInterruptIfRunning)
		                                                    {
			                                                    return false;
		                                                    }
		                                                    
		                                                    @Override
		                                                    public boolean isCancelled()
		                                                    {
			                                                    return false;
		                                                    }
		                                                    
		                                                    @Override
		                                                    public boolean isDone()
		                                                    {
			                                                    return true;
		                                                    }
		                                                    
		                                                    @Override
		                                                    public Void get() throws InterruptedException, ExecutionException
		                                                    {
			                                                    return null;
		                                                    }
		                                                    
		                                                    @Override
		                                                    public Void get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
		                                                    {
			                                                    return null;
		                                                    }
	                                                    };
	
	public AioClient setReadTimeout(long readTimeout)
	{
		this.readTimeout = readTimeout;
		return this;
	}
	
	public AioClient setWaitTimeout(long waitTimeout)
	{
		this.waitTimeout = waitTimeout;
		return this;
	}
	
	public AioClient setAddress(String address)
	{
		this.address = address;
		return this;
	}
	
	public AioClient setPort(int port)
	{
		this.port = port;
		return this;
	}
	
	public AioClient setChannelGroup(AsynchronousChannelGroup channelGroup)
	{
		this.channelGroup = channelGroup;
		return this;
	}
	
	public AioClient setReadHandlers(DataHandler... readHandlers)
	{
		this.readHandlers = readHandlers;
		return this;
	}
	
	public AioClient setWriteHandlers(DataHandler... writeHandlers)
	{
		this.writeHandlers = writeHandlers;
		return this;
	}
	
	public AioClient setFrameDecodec(FrameDecodec frameDecodec)
	{
		this.frameDecodec = frameDecodec;
		return this;
	}
	
	public AioClient connect() throws IOException, InterruptedException, ExecutionException
	{
		if (connectState == UNCONNECTED)
		{
			socketChannel = AsynchronousSocketChannel.open(channelGroup);
			socketChannel.connect(new InetSocketAddress(address, port)).get();
			readCompleter = new ClientReadCompleter(frameDecodec, readHandlers);
			readAndWait();
			connectState = CONNECTED;
		}
		return this;
	}
	
	protected void continueRead()
	{
		socketChannel.read(getReadBuffer(), readTimeout, TimeUnit.MILLISECONDS, this, readCompleter);
	}
	
	protected void readAndWait()
	{
		socketChannel.read(getReadBuffer(), waitTimeout, TimeUnit.MILLISECONDS, this, readCompleter);
		
	}
	
	private ByteBuffer getReadBuffer()
	{
		ioBuf.compact();
		ByteBuffer ioBuffer = ioBuf.nioBuffer();
		ioBuffer.position(ioBuffer.limit()).limit(ioBuffer.capacity());
		return ioBuffer;
	}
	
	/**
	 * 将一个对象写出并且返回一个future。该future表明的是服务端对该请求报文的响应报文的处理结果
	 * 
	 * @param object
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws SocketException
	 */
	public Future<?> write(Object object) throws InterruptedException, ExecutionException, SocketException
	{
		return write(object, 0);
	}
	
	/**
	 * 将一个对象写出并且指定开始处理时的handler顺序，然后返回一个future。该future表明的是服务端对该请求报文的响应报文的处理结果
	 * 
	 * @param object
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws SocketException
	 */
	public Future<?> write(Object data, int index) throws InterruptedException, ExecutionException, SocketException
	{
		if (connectState == UNCONNECTED)
		{
			throw new InterruptedException("链接已经中断，请重新链接后再发送信息");
		}
		ClientInternalResult internalResult = InternalResultContainer.buildClient(data, this, index);
		for (int i = index; i < writeHandlers.length;)
		{
			data = writeHandlers[i].handle(data, internalResult);
			if (i == internalResult.getIndex())
			{
				i++;
				internalResult.setIndex(i);
			}
			else
			{
				i = internalResult.getIndex();
			}
			
		}
		if (data instanceof ByteBuf<?>)
		{
			Future<?> result = buildFuture();
			ByteBuf<?> buf = (ByteBuf<?>) data;
			int writed = 0;
			while (buf.remainRead() > 0)
			{
				writed = socketChannel.write(((ByteBuf<?>) data).nioBuffer()).get();
				buf.addReadIndex(writed);
			}
			buf.release();
			return result;
		}
		else
		{
			return NORESULT;
		}
	}
	
	protected abstract Future<?> buildFuture();
	
	protected abstract void popOneFuture(Object obj);
	
	protected abstract void releaseAllFuture(Throwable e);
	
	public AioClient close(Throwable e)
	{
		try
		{
			connectState = UNCONNECTED;
			socketChannel.close();
			ioBuf.release();
		}
		catch (IOException e1)
		{
			e.printStackTrace();
		}
		releaseAllFuture(e);
		return this;
	}
}
